package com.zhike.function;

import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
import io.debezium.data.Envelope;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

import java.util.List;


/**
 * @Author: zhike
 * @CreateTime: 2024/2/2
 * @Description:
 */
public class CDCDeserialization implements DebeziumDeserializationSchema<String> {
    @Override
    public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {
//1.创建json对象用于存储最终数据
        JSONObject result = new JSONObject();
        //2.获取库名表名
        String topic = sourceRecord.topic();
        String[] split = topic.split("\\.");
        String database = split[1];
        String tableName = split[2];
        Struct value = (Struct) sourceRecord.value();
        //3.获取"before"数据
        Struct before = value.getStruct("before");
        //创建json对象用于存放before的value值
        JSONObject beforeJson = new JSONObject();  //JSON全大写 不是JsonOBject
        if (before != null) {
            Schema beforeSchema = before.schema();
            List<Field> beforeFields = beforeSchema.fields();
            for (Field beforeField : beforeFields) {
                Object o = before.get(beforeField);
                beforeJson.put(beforeField.name(), o);
            }
        }
        //4.获取"after"数据
        Struct after = value.getStruct("after");
        //创建json对象用于存放after的value值
        JSONObject afterJson = new JSONObject();
        if (after != null) {
            Schema afterSchema = after.schema();
            List<Field> afterFields = afterSchema.fields();
            for (Field afterField : afterFields) {
                Object afterValue = after.get(afterField);
                afterJson.put(afterField.name(), afterValue);
            }
        }
        Envelope.Operation operation = Envelope.operationFor(sourceRecord);
        String type = operation.toString().toLowerCase();
//        if (type.equals("create")) {
//            type = "insert";
//        }
        //{"database":"gmall2023",
        // "before":{"tm_name":"23","logo_url":"32325","id":12},
        // "after":{"tm_name":"23","logo_url":"32","id":12},
        // "type":"update",
        // "tableName":"base_trademark"}
        //6.将字段写入json对象
        result.put("database", database);
        result.put("tableName", tableName);
        result.put("before", beforeJson);
        result.put("after", afterJson);
        result.put("type", type);
        //7.输出数据
        collector.collect(result.toJSONString());
    }

    @Override
    public TypeInformation<String> getProducedType() {
        return BasicTypeInfo.STRING_TYPE_INFO;
    }

}
